package com.mjf.spark.day04

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 * 转换算子-aggregateByKey
 *    按照key对分区内以及分区间的数据进行处理
 *    -aggregateByKey(初始值)(分区内计算规则,分区间计算规则)
 *    -zeroValue:给每一个分区中的每一个key一个初始值
 *    -seqOp:用于在每一个分区中用初始值逐步迭代value
 *    -combOp:用于合并每个分区中的结果
 */
object Spark04_Transformation_aggregateByKey {
  def main(args: Array[String]): Unit = {

    // 创建SparkConf配置文件
    val conf = new SparkConf().setMaster("local[*]").setAppName("Spark04_Transformation_aggregateByKey")
    // 创建SparkContext对象
    val sc = new SparkContext(conf)

    val rdd: RDD[(String, Int)] = sc.makeRDD(List(("a", 3), ("a", 2), ("c", 4), ("b", 3), ("c", 6), ("c", 8)), 2)

    // reduceByKey实现WordCount
//    rdd.reduceByKey(_ + _).collect().foreach(println)

    println("-----------------------------------")

    // aggregateByKey实现WordCount
    rdd.aggregateByKey(0)(_ + _, _ + _).collect().foreach(println)

    // 分区最大值求和
    val resRDD: RDD[(String, Int)] = rdd.aggregateByKey(0)(
      (x, y) => math.max(x, y),   // math.max(_, _)
      (a, b) => a + b   // _ + _
    )

    println("-----------------------------------")

    rdd.mapPartitionsWithIndex{
      case (index, datas) =>
        println(index + "--->" + datas.mkString(","))
        datas
    }.collect().foreach(println)

    println("-----------------------------------")
    resRDD.collect().foreach(println)

    // 关闭连接
    sc. stop()

  }
}

